//TODO 优化检查 是否有做好内存回收工作、如果有必要使用shared_ptr来代替传统指针
//TODO 优化 全局变量尽量使用指针类型 防止全局&静态变量存储区 OOM
//TODO 优化 Vector List等容器里的长期变化存储尽量使用指针，防止大量拷贝出现导致效率降低。删除元素视释放并置空指针
#include <condition_variable>
#include <thread>

#include "vas_do.h"
#include "vas_utils.h"
#include "vas_config.h"
#include "vas_algorithm.h"
#include "vas_video.h"
#include "vas_rocketmq.h"
#include "vas_zk.h"
#include "vas_http.h"

using namespace vas;

static AlgorithmPoolData *ALGO_POOL = new AlgorithmPoolData();

void HandleFrame(const cv::Mat &image, const Channel &channel, long timestamp) {
    if (ALGO_POOL->queue.size() > ALGO_POOL->queue_max) {
        std::stringstream msg;
        msg << "Algo pool is max:" << ALGO_POOL->queue_max << " drop frame of " << channel.channel << " at "
            << timestamp;
        VAS_INFO(msg.str());
        return;
    }
    AlgorithmData *data = BuildAlgorithmData(image, channel, timestamp);
    {
        std::lock_guard<std::mutex> lk(ALGO_POOL->mtx);
        ALGO_POOL->queue.emplace(data);
    }
    ALGO_POOL->cond.notify_one();
}

void AlgorithmTheadFunc(AlgorithmPoolData *pool_data, const VasConfig &config_data) {
    std::unique_lock<std::mutex> lk(pool_data->mtx);
    for (;;) {
        if (!pool_data->queue.empty()) {
            auto algorithm_data = pool_data->queue.front();
            pool_data->queue.pop();
            lk.unlock();
            try {
                RouteAlgorithm(algorithm_data, config_data.algorithm_mock);
            }
            catch (std::exception &e) {
                VAS_ERROR("Algo thread Get Exception, thread dead");
                VAS_ERROR(e.what());
                break;
            }
            if (algorithm_data->traces.size() < algorithm_data->routes.size()) {
                lk.lock();
                pool_data->queue.emplace(algorithm_data);
                lk.unlock();
                pool_data->cond.notify_one();
            } else {
                delete algorithm_data;
            }
            lk.lock();
        } else if (pool_data->shutdown) {
            break;
        } else {
            pool_data->cond.wait(lk);
        }
    }
}

void VideoThreadFunc(const Channel &channel) {
    try {
        ReadVideo(channel, HandleFrame);
        UnregisterVideo(channel);
    }
    catch (std::exception &e) {
        VAS_ERROR("Video thread Get Exception, thread dead");
        VAS_ERROR(e.what());
    }
}

int main() {
    std::vector<std::thread> *video_threads = nullptr;
    zhandle_t *zookeeper_handle = nullptr;
    Pistache::Http::Endpoint *http_server = nullptr;
    try {
        // init order: Cache Config Algorithm Video HttpServer
        LoggerInit();
        VAS_INFO(Banner());
        // init cache
        CacheInit();
        // init Config
        VasConfig config = ReadConfig(FileToString("conf.json"));
        // init algorithm and thread pool
        RocketMQInit(config.rocket_mq_host);

        for (auto &i: config.algorithm_mock) {
            if (i.first == FIRE_RECOGNITION_ALGO) {
                FireRecognitionInit(i.second);
            } else if (i.first == HUMAN_DETECTION_ALGO) {
                HumanDetectionInit(i.second);
            }
        }
        VAS_INFO("Start algorithm pool with " + std::to_string(config.algorithm_threads) + " threads");
        for (int i = 0; i < config.algorithm_threads; i++) {
            std::thread(AlgorithmTheadFunc, ALGO_POOL, config).detach();
        }
        // init Http Server
        VAS_INFO("Start http server at " + std::to_string(config.http_port) + " with " +
                 std::to_string(config.http_threads) + " threads");
        http_server = StartHttpServer(config.http_port, config.http_threads);
        // init video and channels
        video_threads = new std::vector<std::thread>();
        VideoInit();
        if (config.zookeeper_host.empty()) {
            VAS_INFO("Load " + std::to_string(config.local_channels.size()) + " channels from local");
            for (auto &local_channel : config.local_channels) {
                while (GetProcessingVideoNum() >= config.max_videos) {
                    std::this_thread::sleep_for(std::chrono::milliseconds(1000));
                }
                try {
                    RegisterVideo(local_channel);
                }
                catch (std::exception &e) {
                    VAS_ERROR(e.what());
                    continue;
                }
                video_threads->emplace_back(std::thread(VideoThreadFunc, local_channel));
            }
        } else {
            bool running = true;
            VAS_INFO("Load channels from zk " + config.zookeeper_host);
            zookeeper_handle = ZookeeperInit(config.zookeeper_host, &running, CHANNEL_ROOT);

            if (zookeeper_handle == nullptr) {
                running = false;
            }

            while (running) {
                std::this_thread::sleep_for(std::chrono::milliseconds(1000));
                if (GetProcessingVideoNum() >= config.max_videos) {
                    continue;
                }
                try {
                    Channel channel = GetOneWaitingChannel(zookeeper_handle, CHANNEL_ROOT, HANDLE_NODE,
                                                           (void *) StopVideo);
                    try {
                        RegisterVideo(channel);
                    }
                    catch (std::exception &e) {
                        VAS_ERROR(e.what());
                        continue;
                    }
                    video_threads->emplace_back(std::thread(VideoThreadFunc, channel));
                }
                catch (VasNotFound &e) {
                    VAS_INFO(e.what());
                }
            }
        }

    }
    catch (std::exception &e) {
        VAS_ERROR("Main Get Exception");
        VAS_ERROR(e.what());
    }

    if (video_threads != nullptr) {
        for (auto &i: *video_threads) {
            i.join();
        }
        delete video_threads;
    }

    if (ALGO_POOL != nullptr) {
        {
            std::lock_guard<std::mutex> lk(ALGO_POOL->mtx);
            ALGO_POOL->shutdown = true;
        }
        ALGO_POOL->cond.notify_all();
    }

    if (http_server != nullptr) {
        http_server->shutdown();
        delete http_server;
    }

    if (zookeeper_handle != nullptr) {
        ZookeeperStop(zookeeper_handle);
    }

    RocketMQStop();
}